package cn.z2huo.demo.kafka.apache.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * <p>
 *
 * @author z2huo
 */
@Slf4j
public class SyncAndAsyncCommitConsumer {

    public static final String TOPIC = "test-topic";

    public static final String GROUP_ID = "my-consumer-group";

    public static void main(String[] args) {
        new SyncAndAsyncCommitConsumer().consume();
    }

    /**
     * <p>手动提交偏移量
     * <p>手动提交也需要指定 group.id，否则会报错 To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.
     */
    public void consume() {
        Properties properties = ConsumerProperties.getConsumerProperties();
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID + "-" + RandomStringUtils.randomNumeric(4));

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        try {
            consumer.subscribe(Collections.singletonList(TOPIC));
            final int minBatchSize = 5;
            List<ConsumerRecord<String, String>> bufferList = new ArrayList<>();
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    bufferList.add(record);
                }
                if (bufferList.size() >= minBatchSize) {
                    for (ConsumerRecord<String, String> record : bufferList) {
                        log.info("Received message, partition is {}, offset is {}, key is {}, value is {}",
                                record.partition(), record.offset(), record.key(), record.value());
                    }
                    consumer.commitSync();
                    bufferList.clear();
                }
            }
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
}
